package com.atguigu.bigdata.gmall.realtime.app.dwd;

import com.atguigu.bigdata.gmall.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.gmall.realtime.common.Constant;
import com.atguigu.bigdata.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2022/10/11 14:33
 */
public class Dwd_03_DwdTradeOrderDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_03_DwdTradeOrderDetail().init(
            3003,
            2,
            "Dwd_03_DwdTradeOrderDetail"
        );
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        
        // 1. 读取 ods_db
        readOdsDb(tEnv, "Dwd_03_DwdTradeOrderDetail");
        // 2. 读取 字典表
        readBaseDic(tEnv);
        // 3. 过滤出详情表
        Table orderDetail = tEnv.sqlQuery("select " +
                                              "data['id'] id,  " +
                                              "data['order_id'] order_id,  " +
                                              "data['sku_id'] sku_id,  " +
                                              "data['sku_name'] sku_name,  " +
                                              "data['create_time'] create_time,  " +
                                              "data['source_id'] source_id,  " +
                                              "data['source_type'] source_type,  " +
                                              "data['sku_num'] sku_num,  " +
                                              "cast( cast(data['sku_num'] as decimal(16,2)) * " +
                                              "      cast(data['order_price'] as decimal(16,2)) as String) split_original_amount,  " +
                                              "data['split_total_amount'] split_total_amount,  " +
                                              "data['split_activity_amount'] split_activity_amount,  " +
                                              "data['split_coupon_amount'] split_coupon_amount,  " +
                                              "ts,  " +
                                              "pt  " +
                                              "from ods_db " +
                                              "where `database`='gmall2022' " +
                                              "and `table`='order_detail' " +
                                              "and `type`='insert' ");
        tEnv.createTemporaryView("order_detail", orderDetail);
        
        // 4. 过滤订单表
        Table orderInfo = tEnv.sqlQuery("select " +
                                            "data['id'] id," +
                                            "data['user_id'] user_id," +
                                            "data['province_id'] province_id " +
                                            "from ods_db " +
                                            "where `database`='gmall2022' " +
                                            "and `table`='order_info' " +
                                            "and `type`='insert' ");
        tEnv.createTemporaryView("order_info", orderInfo);
        // 5. 过滤出详情活动表
        Table orderDetailActivity = tEnv.sqlQuery("select " +
                                                      "data['order_detail_id'] order_detail_id," +
                                                      "data['activity_id'] activity_id," +
                                                      "data['activity_rule_id'] activity_rule_id " +
                                                      "from ods_db " +
                                                      "where `database`='gmall2022' " +
                                                      "and `table`='order_detail_activity' " +
                                                      "and `type`='insert' ");
        tEnv.createTemporaryView("order_detail_activity", orderDetailActivity);
        // 6. 过滤出详情优惠券表
        Table orderDetailCoupon = tEnv.sqlQuery("select  " +
                                                    "data['order_detail_id'] order_detail_id,  " +
                                                    "data['coupon_id'] coupon_id  " +
                                                    "from `ods_db`  " +
                                                    "where `database`='gmall2022' " +
                                                    "and `table` = 'order_detail_coupon'  " +
                                                    "and `type` = 'insert'  ");
        tEnv.createTemporaryView("order_detail_coupon", orderDetailCoupon);
        // 7. 5 张表的 join
        Table result = tEnv.sqlQuery("select " +
                                         "od.id,  " +
                                         "od.order_id,  " +
                                         "oi.user_id,  " +
                                         "od.sku_id,  " +
                                         "od.sku_name,  " +
                                         "oi.province_id,  " +
                                         "act.activity_id,  " +
                                         "act.activity_rule_id,  " +
                                         "cou.coupon_id,  " +
                                         "date_format(od.create_time, 'yyyy-MM-dd') date_id,  " +
                                         "od.create_time,  " +
                                         "od.source_id,  " +
                                         "od.source_type,  " +
                                         "dic.dic_name source_type_name,  " +
                                         "od.sku_num,  " +
                                         "od.split_original_amount,  " +
                                         "od.split_activity_amount,  " +
                                         "od.split_coupon_amount,  " +
                                         "od.split_total_amount,  " +
                                         "od.ts, " +
                                         "current_row_timestamp() row_op_ts  " + /*得到每行计算的时间: 到后面 dws 去重的时候使用*/
                                         "from order_detail od " +
                                         "join order_info oi on od.order_id=oi.id " +
                                         "left join order_detail_activity act on od.id=act.order_detail_id " +
                                         "left join order_detail_coupon cou on od.id=cou.order_detail_id " +
                                         "join base_dic for system_time as of od.pt as dic on od.source_type=dic.dic_code ");
        
        
        // 7. 写出到 kafka 中
        tEnv.executeSql("create table dwd_trade_order_detail(  " +
                            "id string,  " +
                            "order_id string,  " +
                            "user_id string,  " +
                            "sku_id string,  " +
                            "sku_name string,  " +
                            "province_id string,  " +
                            "activity_id string,  " +
                            "activity_rule_id string,  " +
                            "coupon_id string,  " +
                            "date_id string,  " +
                            "create_time string,  " +
                            "source_id string,  " +
                            "source_type string,  " +
                            "source_type_name string,  " +
                            "sku_num string,  " +
                            "split_original_amount string,  " +
                            "split_activity_amount string,  " +
                            "split_coupon_amount string,  " +
                            "split_total_amount string,  " +
                            "ts bigint,  " +
                            "row_op_ts timestamp_ltz(3),  " +
                            "primary key(id) not enforced  " +
                            ")" + SQLUtil.getUpsertKafkaSink(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));
    
        result.executeInsert("dwd_trade_order_detail");
    }
}
/*
详情 id   分摊总金额    活动 id     优惠券 id
1000        1222      null          null          1
        null
1000        1222      1             null          2
1000        1222      1             2             3

由于 left join 的存在, 导致同一个详情的数据会重复写入到 kafka 中, 消费的时候重复消费

计算总金额有些数据会被重复计算

dws 层汇总的时候, 需要去重!!!
    保留最全的.
        一定是最晚计算出来.
        key 在每条数据数据中生成一个时间戳, 时间戳最大的那个就是数据最全的.




----------------------------
下单:
  order_info
    insert 一条数据
      insert
  order_detail
    insert 多条多数据 粒度是 sku
    只要 sku 相关的信息, 需要 省份, 用户信息 ...
      insert

  order_detail_activity
    详情参数的活动
      insert
  order_detail_coupon
    详情使用的优惠券信息
      insert
  base_dic
    字典表, 用来退化维度


order_detail
  join  on 订单 id
order_info
  left join on 详情 id
order_detail_activity
  left join on 详情 id
order_detail_coupon
  lookup join
base_dic
 */